package com.yeming.flink.practice.cep

import java.text.SimpleDateFormat
import java.util

import org.apache.flink.cep.PatternSelectFunction
import org.apache.flink.cep.scala.{CEP, PatternStream}
import org.apache.flink.cep.scala.pattern.Pattern
import org.apache.flink.streaming.api.TimeCharacteristic
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.windowing.time.Time


/**
 * 登录日志样例类
 *
 * @param id 登录日志ID
 * @param userName 登录用户名
 * @param eventType 登录类型
 * @param eventTime 登录发生时间 精确到秒
 */
case class LoginEvent(id:Long, userName:String, eventType:String, eventTime:Long)

object TestCepByLogin {

  //从一堆的登录日志中，匹配一个恶意登录的模式（如果一个用户在十秒内连续登录失败三次，则是恶意登录），从而找到那些用户名是恶意登录
  def main(args: Array[String]): Unit = {

    val streamEnv: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    streamEnv.setParallelism(1)
    streamEnv.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)
    val stream: DataStream[LoginEvent] = streamEnv.fromCollection(List(
      new LoginEvent(1, "张三", "fail", 1577080457),
      new LoginEvent(2, "张三", "fail", 1577080458),
      new LoginEvent(3, "张三", "fail", 1577080459),
      new LoginEvent(4, "李四", "fail", 1577080460),
      new LoginEvent(5, "李四", "success", 1577080462),
      new LoginEvent(6, "张三", "fail", 1577080462)
    )).assignAscendingTimestamps(_.eventTime * 1000L)

    //定义pattern
    val pattern: Pattern[LoginEvent, LoginEvent] = Pattern.begin[LoginEvent]("start").where(_.eventType.equals("fail"))
      .next("failnext").where(_.eventType.equals("fail"))
      .next("failAgainNext").where(_.eventType.equals("fail"))
      .within(Time.seconds(10))//时间限制，基于事件时间
    //检测模式
    val patternStream: PatternStream[LoginEvent] = CEP.pattern(stream.keyBy(_.userName), pattern)// 根据用户名进行分组

    //选择结果并输出
    val result: DataStream[String] = patternStream.select(new PatternSelectFunction[LoginEvent,String] {
      override def select(map: util.Map[String, util.List[LoginEvent]]): String = {
//        val format = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss")
        val keyIter: util.Iterator[String] = map.keySet().iterator()
        val e1: LoginEvent = map.get(keyIter.next()).get(0)
        val e2: LoginEvent = map.get(keyIter.next()).get(0)
        val e3: LoginEvent = map.get(keyIter.next()).get(0)
        String.format("用户登录名称：%s，用户第一次登陆时间：%s，用户第二次登录时间：%s，用户第三次登录时间：%s。", e1.userName
          , e1.eventTime.toString, e2.eventTime.toString, e3.eventTime.toString)
      }
    })

    result.print("匹配到的用户")
    streamEnv.execute()

  }

}
